package Kafka;

import org.apache.commons.lang.RandomStringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.text.DecimalFormat;
import java.util.Properties;
import java.util.Random;

import static java.lang.Double.parseDouble;

public class CreateOne {
    public static void main(String[] args) {

        // 1. 创建 kafka 生产者的配置对象
        Properties properties = new Properties();
        // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "niit01:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");


        // 每隔5秒推送 5条数据
        while (true) {
            // 3. 创建 kafka 生产者对象
            KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
            System.out.println("开始发送数据");
            try {
                // 4. 调用 send 方法,发送消息，一次发送5条
                for (int i = 0; i <5; i++) {
                    Random rd = new Random();  //创建实例
                    //班级号  范围在202001-202003之间
                    int classNo = rd.nextInt(3)+202001;
                    //名字 从char数组中随机挑选5个字符组成
                    String str=RandomStringUtils.random(5,new char[]{'a','b','c','d','e','f','g','h','i','j','k','l','m','n'});
                    //学期 1-8
                    int semester = rd.nextInt(8)+1;
                    // 0 1性别
                    int gender = rd.nextInt(2);
                    //出生日期 从gender数组中随机挑选
                    String[] birth = new String[]{"20020318","20010924","20030817","20020320","20020908","20020219","20020506","20020729"};
                    int bir=rd.nextInt(8);
                    String B = birth[bir];
                    //学号 2020340701-2020340799
                    int StudId = rd.nextInt(99) +2020340701;
                    //学籍状态 随机选择"L"或"A"
                    String[] Statue = new String[]{"L", "A"};
                    int Stu = rd.nextInt(2);
                    String S = Statue[Stu];
                    //绩点
                    DecimalFormat decimalFormat = new DecimalFormat("0.00");  //规定绩点的格式--》必须保留两位小数，不够的后补0
                    double GPA = parseDouble(decimalFormat.format(rd.nextDouble()*3.0+2.0));  //生成[2.0-5.0]区间的小数 保留两位，强制类型转换
                    // 品类
                    Object[] os = new Object[8];  //Object数组可包含不同类型内容
                    os[0] = classNo;
                    os[1] = str;
                    os[2] = gender;
                    os[3] = B;
                    os[4] = StudId;
                    os[5] = semester;
                    os[6] = GPA;
                    os[7] = S;
//                    直接造一整条数据，随机选取整条数据发送
//                    String[] categoryArray = new String[]{"202006\tCrystal\t0\t20020320\t202034070726\t5\t4.01\tA", "202005\tVictoria\t0\t20020319\t202034070728\t5\t3.86\tA"};
//                    int radom =rd.nextInt(2);
//                    String category  = categoryArray[radom];
                    // 模拟数据
                    System.out.println("发送数据为" + os[0]+"\t"+os[1]+"\t"+os[2]+"\t"+os[3]+"\t"+os[4]+"\t"+os[5]+"\t"+os[6]+"\t"+os[7]);

                    try {

                        kafkaProducer.send(new ProducerRecord<String, String>("stuInfo1", os[0]+"\t"+os[1]+"\t"+os[2]+"\t"+os[3]+"\t"+os[4]+"\t"+os[5]+"\t"+os[6]+"\t"+os[7]));
//                        kafkaProducer.send(new ProducerRecord<String, String>("test", category));

                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
                Thread.sleep(5000);
            }catch(InterruptedException e){
                e.printStackTrace();
            }


//             5. 关闭资源
            kafkaProducer.close();

        }

    }
}